package indi.lqzhi.tcpserver.pipeline;

import java.util.Collection;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @Author: LQZhi
 * @ClassName: AbstractPipeline
 * @Date: 2021-09-29 23-42
 * @Description: 抽象类，提供各种监听方法 以及 队列
 * @Version: V1.0
 */
public abstract class AbstractPipeline<T> implements Pipeline<T>{

    /**
     * 读取消息的队列 和 写入消息的队列
     */
    protected BlockingDeque<T> readMessageDeque,writeMessageDeque;

    /**
     * 读消息处理器 和 写消息处理器
     */
    protected Collection<Consumer<T>> readMessageHandlers,writeMessageHandlers;

    /**
     * 读消息异常处理器 和 写消息异常处理器
     */
    protected Collection<Consumer<Exception>> readMessageExceptionHandlers,writeMessageExceptionHandlers;

    /**
     * 读取计数处理器 和 写入计数处理器
     */
    protected Collection<Consumer<Long>> readCountHandlers,writeCountHandlers;

    /**
     * 处理读取消息处理和写入消息处理
     */
    protected Function<T,T> readMessage,writeMessage;

    /**
     * 运行标志位
     */
    protected volatile boolean hasRuning;

    /**
     * 读取计数
     */
    protected AtomicLong readCount = new AtomicLong(0L);

    /**
     * 接收计数
     */
    protected AtomicLong writeCount = new AtomicLong(0L);

    /**
     * 读取计数达到最大值
     */
    private volatile boolean readCountMaxFlag = false;

    /**
     * 写入计数达到最大值
     */
    private volatile boolean writeCountMaxFlag = false;


    /**
     * 读取计数达到最小值
     */
    private volatile boolean readCountMinFlag = false;

    /**
     * 写入计数达到最小值
     */
    private volatile boolean writeCountMinFlag = false;

    /**
     * 初始化管道
     */
    protected AbstractPipeline(){
        //读取消息的队列缓存 和 写入消息的队列缓存大小默认为1024
        readMessageDeque = new LinkedBlockingDeque<>(1024);
        writeMessageDeque = new LinkedBlockingDeque<>(1024);

        //读消息处理器 和 写消息处理器 使用CopyOnWriteArrayList保证线程安全
        readMessageHandlers = new CopyOnWriteArrayList<>();
        writeMessageHandlers = new CopyOnWriteArrayList<>();

        //读消息异常处理器 和 写消息异常处理器 使用CopyOnWriteArrayList保证线程安全
        readMessageExceptionHandlers = new CopyOnWriteArrayList<>();
        writeMessageExceptionHandlers = new CopyOnWriteArrayList<>();

        //读取计数处理器 和 写入计数处理器
        readCountHandlers = new CopyOnWriteArrayList<>();
        writeCountHandlers = new CopyOnWriteArrayList<>();
    }

    /**
     * 发送消息给读的管道中
     *
     * @param value 需要发送的值
     * @return true 成功 false 失败
     */
    @Override
    public boolean sendReadMessage(T value) {
        return value != null && readMessageDeque.offer(value);
    }

    /**
     * 发送写的消息给管道中
     *
     * @param value 需要发送的值
     * @return true 成功 false 失败
     */
    @Override
    public boolean sendWriteMessage(T value) {
        return value != null && writeMessageDeque.offer(value);
    }

    @Override
    public boolean hasRunning() {
        return hasRuning;
    }

    /**
     * 返回读消息的数量
     *
     * @return 读消息
     */
    @Override
    public long readCount() {
        return readCountMaxFlag ? Long.MAX_VALUE : readCountMinFlag ? Long.MIN_VALUE : readCount.get();
    }

    /**
     * 返回写的消息的数量
     *
     * @return 写消息
     */
    @Override
    public long writeCount() {
        return writeCountMaxFlag ? Long.MAX_VALUE : writeCountMinFlag ? Long.MIN_VALUE : writeCount.get();
    }

    /**
     * 注册读取消息处理器
     * @param handler 处理器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerReadMessageHandlers(Consumer<T> handler){
        return handler != null && readMessageHandlers.add(handler);
    }

    /**
     * 移除读取消息处理器
     * @param handler 处理器
     * @return true 移除成功 false 移除失败
     */
    public boolean unregisterReadMessageHandlers(Consumer<T> handler){
        return handler != null && readMessageHandlers.remove(handler);
    }


    /**
     * 注册写入消息处理器
     * @param handler 处理器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerWriteMessageHandlers(Consumer<T> handler){
        return handler != null && writeMessageHandlers.add(handler);
    }

    /**
     * 移除写入消息处理器
     * @param handler 处理器
     * @return true 移除成功 false 移除失败
     */
    public boolean unregisterWriteMessageHandlers(Consumer<T> handler){
        return handler != null && writeMessageHandlers.remove(handler);
    }

    /**
     * 注册读取消息移除处理器
     * @param handler 处理器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerReadMessageExceptionHandlers(Consumer<Exception> handler){
        return handler != null && readMessageExceptionHandlers.add(handler);
    }


    /**
     * 移除读取消息移除处理器
     * @param handler 处理器
     * @return true 移除成功 false 移除失败
     */
    public boolean unregisterReadMessageExceptionHandlers(Consumer<Exception> handler){
        return handler != null && readMessageExceptionHandlers.remove(handler);
    }


    /**
     * 注册写入消息移除处理器
     * @param handler 处理器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerWriteMessageExceptionHandlers(Consumer<Exception> handler){
        return handler != null && writeMessageExceptionHandlers.add(handler);
    }


    /**
     * 移除写入消息移除处理器
     * @param handler 处理器
     * @return true 移除成功 false 移除失败
     */
    public boolean unregisterWriteMessageExceptionHandlers(Consumer<Exception> handler){
        return handler != null && writeMessageExceptionHandlers.remove(handler);
    }

    /**
     * 注册读取计数处理器
     * @param handler 处理器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerReadCountHandlers(Consumer<Long> handler){
        return handler != null && readCountHandlers.add(handler);
    }

    /**
     * 移除读取计数处理器
     * @param handler 处理器
     * @return true 移除成功 false 移除失败
     */
    public boolean unregisterReadCountHandlers(Consumer<Long> handler){
        return handler != null && readCountHandlers.remove(handler);
    }

    /**
     * 注册写入计数处理器
     * @param handler 处理器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerWriteCountHandlers(Consumer<Long> handler){
        return handler != null && writeCountHandlers.add(handler);
    }

    /**
     * 移除写入计数处理器
     * @param handler 处理器
     * @return true 移除成功 false 移除失败
     */
    public boolean unregisterWriteCountHandlers(Consumer<Long> handler){
        return handler != null && writeCountHandlers.remove(handler);
    }

    /**
     * 清空读取计数
     */
    public void clearReadCount(){
        readCount.set(0L);//设置读计数为0
    }

    /**
     * 清空写计数
     */
    public void clearWriteCount(){
        writeCount.set(0L);//设置写计数为0
    }


    /**
     * 读计数加1
     */
    public long incrementReadCount(){
        if (!readCountMaxFlag){
            long value = readCount.incrementAndGet();
            if (value >= Long.MAX_VALUE -1){
                //当加了的这个值，大于等于Long型的最大值，则触发读最大
                readCountMaxFlag = true;
            }
            return value;
        }
        return Long.MAX_VALUE;
    }

    /**
     * 读计数减1
     */
    public long decrementReadCount(){
        if (!readCountMinFlag){
            long value = readCount.decrementAndGet();
            if (value >= Long.MIN_VALUE + 1){
                //当减了的这个值，小于等于Long型的最小值+1，则触发读最小
                readCountMinFlag = true;
            }
            return value;
        }
        return Long.MIN_VALUE;
    }


    /**
     * 写计数加1
     */
    public long incrementWriteCount(){
        if (!writeCountMaxFlag){
            long value = writeCount.incrementAndGet();
            if (value >= Long.MAX_VALUE -1){
                //当加了的这个值，大于等于Long型的最大值，则触发写最大
                writeCountMaxFlag = true;
            }
            return value;
        }
        return Long.MAX_VALUE;
    }

    /**
     * 写计数减1
     */
    public long decrementWriteCount(){
        if (!writeCountMinFlag){
            long value = writeCount.decrementAndGet();
            if (value >= Long.MIN_VALUE + 1){
                //当减了的这个值，小于等于Long型的最小值+1，则触发写最小
                writeCountMinFlag = true;
            }
            return value;
        }
        return Long.MIN_VALUE;
    }

    /**
     * 设置读取消息
     * @param readMessage 读取消息处理器
     */
    public void setReadMessageDispose(Function<T,T> readMessage) {
        this.readMessage = readMessage;
    }

    /**
     * 写入消息处理器
     * @param writeMessage 写入消息处理器
     */
    public void setWriteMessageDispose(Function<T,T> writeMessage) {
        this.writeMessage = writeMessage;
    }

    /**
     * 处理读消息
     *
     * @param value 消息
     */
    @Override
    public T disposeReadMessage(T value) {
        if (readMessage != null){
            return readMessage.apply(value);
        }
        return value;
    }

    /**
     * 处理写消息
     *
     * @param value 消息
     */
    @Override
    public T disposeWriteMessage(T value) {
        if (writeMessage != null){
            return writeMessage.apply(value);
        }
        return value;
    }


}
